package com.kitten.note.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.google.common.collect.Lists;
import com.kitten.framework.common.utils.JsonUtils;
import com.kitten.note.biz.constants.MQConstants;
import com.kitten.note.biz.domain.dataobject.NoteLikeDO;
import com.kitten.note.biz.domain.mapper.NoteLikeDOMapper;
import com.kitten.note.biz.model.dto.LikeUnLikeNoteMqDTO;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author kitten
 */
@Slf4j
@Component
public class BatchLikeUnLikeNoteConsumer {
    @Resource
    private NoteLikeDOMapper noteLikeDOMapper;

    @Value("${rocketmq.name-server}")
    private String nameServer;

    private DefaultMQPushConsumer consumer;

    private RateLimiter rateLimiter = RateLimiter.create(10);

    @Bean(name = "BatchLikeUnLikeNoteConsumer")
    public DefaultMQPushConsumer mqPushConsumer() throws MQClientException {
        String group = "kittenBook_record_group_" + MQConstants.TOPIC_LIKE_OR_UNLIKE;
        consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MQConstants.TOPIC_LIKE_OR_UNLIKE, "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setMaxReconsumeTimes(3);    // 最大重试次数
        consumer.setConsumeMessageBatchMaxSize(30);    // 一次拉取最大数量
        consumer.setPullInterval(1000); // 拉取间隔
        // 注册
        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            log.info(" 笔记点赞/取消点赞 本批次消费消息大小: {}", msgs.size());

            try {
                rateLimiter.acquire(); // 限流

                // 消息体 Json 字符串转 DTO
                List<LikeUnLikeNoteMqDTO> likeUnlikeNoteMqDTOS = Lists.newArrayList();
                msgs.forEach(msg -> {
                    String msgJson = new String(msg.getBody());
                    log.info("==> Consumer - Received message: {}", msgJson);
                    likeUnlikeNoteMqDTOS.add(JsonUtils.parseObject(msgJson, LikeUnLikeNoteMqDTO.class));
                });
                // 操作合并
                // 1. 按发起点赞的用户id分组  { 2: [o1, o2, o3], 3: [o4, o5, o6] }
                Map<Long, List<LikeUnLikeNoteMqDTO>> userMap = likeUnlikeNoteMqDTOS.stream()
                        .collect(Collectors.groupingBy(LikeUnLikeNoteMqDTO::getUserId));
                // 2. 按笔记id二次分组
                List<LikeUnLikeNoteMqDTO> finalOperations = userMap.values().stream().flatMap(userOperations -> {
                    // noteGroupMap { noteId1: [o1, o3], noteId2: [o2] }
                    Map<Long, List<LikeUnLikeNoteMqDTO>> noteGroupMap = userOperations.stream().collect(Collectors.groupingBy(LikeUnLikeNoteMqDTO::getNoteId));

                    // 3. 处理每个noteId的分组, 即某个用户对某个笔记的点赞/取消点赞操作
                    Stream<LikeUnLikeNoteMqDTO> userNoteLikeOperationStream = noteGroupMap.entrySet().stream().filter(entry -> {
                        List<LikeUnLikeNoteMqDTO> operations = entry.getValue();
                        int size = operations.size();   // 操作次数
                        if (size % 2 == 0) {
                            return false;
                        } else {
                            return true;
                        }
                    }).map(entry -> {
                        List<LikeUnLikeNoteMqDTO> operations = entry.getValue();
                        return operations.get(operations.size() - 1);
                    });
                    return userNoteLikeOperationStream;
                }).toList();   // finaOperations: [o2]

                // 4. 点赞/取消点赞操作写库
                if (CollUtil.isNotEmpty(finalOperations)) {
                    List<NoteLikeDO> noteLikeDOS = finalOperations.stream().map(finalOperation -> {
                        return NoteLikeDO.builder()
                                .userId(finalOperation.getUserId())
                                .noteId(finalOperation.getNoteId())
                                .createTime(finalOperation.getCreateTime())
                                .status(finalOperation.getType())
                                .build();
                    }).toList();
                    noteLikeDOMapper.batchInsertOrUpdate(noteLikeDOS);
                }

                return ConsumeOrderlyStatus.SUCCESS;
            } catch (Exception e) {
                log.error("", e);
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        });
        consumer.start();
        return consumer;
    }

    @PreDestroy
    public void destroy() {
        if (Objects.nonNull(consumer)) {
            try {
                consumer.shutdown();  // 关闭消费者
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }


}
